package com.java.springcloud.config;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.java.springcloud.listener.ChangeStreamDocumentListener;
import com.mongodb.client.model.changestream.FullDocument;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.MongoTransactionManager;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.Subscription;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.ErrorHandler;

import java.util.concurrent.*;

import static org.springframework.data.mongodb.core.aggregation.Aggregation.match;
import static org.springframework.data.mongodb.core.aggregation.Aggregation.newAggregation;
import static org.springframework.data.mongodb.core.query.Criteria.where;

/**
 * @author dushougudu
 */
@Configuration
@Slf4j
public class MongoConfig{

    @Autowired
    private RedisTemplate redisTemplate;

    private volatile Subscription subscription;

    @Bean
    MessageListenerContainer messageListenerContainer(MongoTemplate template) {
        Executor executor = new ThreadPoolExecutor(4, 4,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(1),
                new ThreadFactoryBuilder().setNameFormat("mongo-message-listener-%d").build());
        MessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer(template, executor) {
            @Override
            public boolean isAutoStartup() {
                return true;
            }
        };
        return messageListenerContainer;
    }


    @Bean(initMethod = "start", destroyMethod = "close")
    LeaderLatch leaderSelectorClient(CuratorFramework curatorFramework,
                                     MessageListenerContainer messageListenerContainer,
                                     ChangeStreamDocumentListener documentMessageListener) {
        LeaderLatch leaderLatch = new LeaderLatch(curatorFramework, "/change_stream_leader");
        leaderLatch.addListener(new LeaderLatchListener() {
            @Override
            public void isLeader() {
                synchronized (this) {
                    if (leaderLatch.hasLeadership() && subscription == null) {
                        ChangeStreamRequest.ChangeStreamRequestBuilder builder =
                                ChangeStreamRequest.builder(documentMessageListener);
                        if (redisTemplate.hasKey("resumeToken")) {
                            Document resumeToken = Document.parse(redisTemplate.opsForValue().get("resumeToken").toString());
                            builder.startAfter(resumeToken.toBsonDocument());
                        }
                        ChangeStreamRequest<Document> request = builder
                                .filter(newAggregation(match(where("operationType").in("insert", "delete", "update", "replace"))))
                                .fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
                                .build();
                        subscription = messageListenerContainer.register(request, Document.class);
                        log.info("成为leader，启动监听器");
                    }
                }
            }

            @Override
            public void notLeader() {
                synchronized (this) {
                    if (!leaderLatch.hasLeadership() && subscription != null) {
                        messageListenerContainer.remove(subscription);
                        subscription = null;
                        log.info("退出leader，关闭监听器");
                    }
                }
            }
        });

        return leaderLatch;
    }


    @Bean
    MongoTransactionManager transactionManager(MongoDatabaseFactory factory) {
        return new MongoTransactionManager(factory);
    }

}
